import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.InputStreamReader;
import java.util.GregorianCalendar;

public class TarefaComQuatroMaquinas22 {
	private static File diretorioBroker = new File("/home/isabellylr/broker");
	private static String diretorioArquivos = "/home/isabellylr/workspace/Twitter";
	private static int INDICE_FILTRAGEM_1, INDICE_FILTRAGEM_2, INDICE_CONTAGEM_1, INDICE_CONTAGEM_2;
	private static int PROXIMO_ARQUIVO = 1;
	private static int PROXIMO_ARQUIVO_FILTRADO = 1;
	private static int QUANTIDADE_DE_ARQUIVOS = 12;
	private static long inicio;
	
	public static void main(String[] args) {
		try {
			inicio = new GregorianCalendar().getTimeInMillis();
			executarTarefas();
			imprimeTempo();
		} catch (Exception ex) {
			System.err.println(ex.getMessage());
		}
	}
	
	private static void executarTarefas() throws Exception {
		System.out.println("---");
		if (terminouJob("FiltroUm" + INDICE_FILTRAGEM_1) && (new File("Files/Tweets" + PROXIMO_ARQUIVO + ".txt")).exists()) {
			INDICE_FILTRAGEM_1++;
			criaFiltroJDF("FiltroUm" + INDICE_FILTRAGEM_1);
			executaJob("FiltroUm" + INDICE_FILTRAGEM_1);
			PROXIMO_ARQUIVO++;
		} if ((terminouJob("FiltroDois" + INDICE_FILTRAGEM_2))  && (new File("Files/Tweets" + PROXIMO_ARQUIVO + ".txt")).exists()) {
			INDICE_FILTRAGEM_2++;
			criaFiltroJDF("FiltroDois" + INDICE_FILTRAGEM_2);
			executaJob("FiltroDois" + INDICE_FILTRAGEM_2);
			PROXIMO_ARQUIVO++;
		} if (terminouJob("ContadorDeOcorrenciaUm" + INDICE_CONTAGEM_1) && (new File("Files/TweetsFiltrados" + PROXIMO_ARQUIVO_FILTRADO + ".txt")).exists()) {
			INDICE_CONTAGEM_1++;
			criaContadorDeOcorrenciaJDF("ContadorDeOcorrenciaUm" + INDICE_CONTAGEM_1);
			executaJob("ContadorDeOcorrenciaUm" + INDICE_CONTAGEM_1);
			PROXIMO_ARQUIVO_FILTRADO++;
		} if (terminouJob("ContadorDeOcorrenciaDois" + INDICE_CONTAGEM_2) && (new File("Files/TweetsFiltrados" + PROXIMO_ARQUIVO_FILTRADO + ".txt")).exists()) {
			INDICE_CONTAGEM_2++;
			criaContadorDeOcorrenciaJDF("ContadorDeOcorrenciaDois" + INDICE_CONTAGEM_2);
			executaJob("ContadorDeOcorrenciaDois" + INDICE_CONTAGEM_2);
			PROXIMO_ARQUIVO_FILTRADO++;
		} if (PROXIMO_ARQUIVO_FILTRADO <= QUANTIDADE_DE_ARQUIVOS ) {
			System.out.println("---");
			System.out.println("Tempo:" + (new GregorianCalendar().getTimeInMillis() - inicio));
			executarTarefas();
		}
	}
	
	private static void criaFiltroJDF(String job) throws Exception {
		FileOutputStream filtroJDF = new FileOutputStream("Jobs/" + job + ".jdf");
		filtroJDF.write(("job :\nlabel 	: " + job + "\n\ntask :\n").getBytes());
		filtroJDF.write(("init 	: put " + diretorioArquivos + "/Files/Tweets" + PROXIMO_ARQUIVO + ".txt Tweets" + PROXIMO_ARQUIVO + ".txt\n").getBytes());
		filtroJDF.write(("put " + diretorioArquivos + "/bin/Filtro.class Filtro.class\n").getBytes());
		filtroJDF.write(("remote 	: java -cp . Filtro Tweets" + PROXIMO_ARQUIVO + ".txt TweetsFiltrados" + PROXIMO_ARQUIVO + ".txt\n").getBytes());
		filtroJDF.write(("final   : get TweetsFiltrados" + PROXIMO_ARQUIVO + ".txt " + diretorioArquivos + "/Files/TweetsFiltrados" + PROXIMO_ARQUIVO + ".txt").getBytes());
	}
	
	private static void criaContadorDeOcorrenciaJDF(String job) throws Exception {
		FileOutputStream ocorrenciaJDF = new FileOutputStream("Jobs/" + job + ".jdf");
		ocorrenciaJDF.write(("job :\nlabel 	: " + job + "\n\ntask :\n").getBytes());
		ocorrenciaJDF.write(("init 	: put " + diretorioArquivos + "/Files/TweetsFiltrados" + PROXIMO_ARQUIVO_FILTRADO + ".txt TweetsFiltrados" + PROXIMO_ARQUIVO_FILTRADO + ".txt\n").getBytes());
		ocorrenciaJDF.write(("put " + diretorioArquivos + "/bin/ContadorDeOcorrencia.class ContadorDeOcorrencia.class\n").getBytes());
		ocorrenciaJDF.write(("remote 	: java -cp . ContadorDeOcorrencia TweetsFiltrados" + PROXIMO_ARQUIVO_FILTRADO + ".txt arquivoOcorrencias" + PROXIMO_ARQUIVO_FILTRADO + ".txt\n").getBytes());
		ocorrenciaJDF.write(("final   : get arquivoOcorrencias" + PROXIMO_ARQUIVO_FILTRADO + ".txt " + diretorioArquivos + "/Files/arquivoOcorrencias" + PROXIMO_ARQUIVO_FILTRADO + ".txt").getBytes());
	}
	
	private static void executaJob(String job) throws Exception {
		imprime(Runtime.getRuntime().exec("bash broker addjob " + diretorioArquivos + "/Jobs/" + job + ".jdf", null, diretorioBroker));
	}
	
	private static boolean terminouJob(String job) throws Exception {
		Process p = Runtime.getRuntime().exec("bash broker status", null, diretorioBroker);
		BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
		String linha = "";
		while ((linha = reader.readLine()) != null) {
			if (linha.contains(job)) {
				return linha.contains("[FINISHED]");
			}
		}
		return true;
	}
	
	private static void imprimeTempo() throws Exception {
		if (!terminouTodosOsJobs()) {
			imprimeTempo();
		} else {
			System.out.println("Tempo total da aplicacao (em milisegundos):" + (new GregorianCalendar().getTimeInMillis() - inicio));
		}
	}
	
	private static boolean terminouTodosOsJobs() throws Exception {
		Process p = Runtime.getRuntime().exec("bash broker status", null, diretorioBroker);
		BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
		String linha = "";
		while ((linha = reader.readLine()) != null) {
			if (linha.contains("[RUNNING]")) {
				return false;
			}
		}
		return true;
	}
	
	private static void imprime(Process p) throws Exception {
		BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
		String linha = "";
		while ((linha = reader.readLine()) != null) {
			System.out.println(linha);
		}
	}
}